package com.ylr.job.server;

/**
 * @author ylr
 * @version 1.0.0
 * @ClassName GrpcServer.java
 * @Description TODO
 * @createTime 2021年11月04日 09:53:00
 */

import com.ylr.job.biz.ExecutorBiz;
import com.ylr.job.biz.impl.ExecutorBizIProtoImpl;
import com.ylr.job.biz.impl.ExecutorBizImpl;
import com.ylr.job.thread.ExecutorRegistryThread;
import io.grpc.Server;
import io.grpc.ServerBuilder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.concurrent.*;


/**
 * @author ylr
 * @version 1.0.0
 * @ClassName GrpcServer.java
 * @Description TODO   gRPC服务启动类,启动时注册添加需要对外提供的服务类
 * @createTime 2021年11月04日 13:01:00
 */
public class GrpcServer {
    private Server server;
    private Thread thread;
    private ExecutorBiz executorBiz;
    private static final Logger logger = LoggerFactory.getLogger(GrpcServer.class);

    public void start(final String address, final int port, final String appname, final String accessToken) {
        executorBiz = new ExecutorBizImpl();
        thread = new Thread(()-> {
            try {

                ThreadPoolExecutor bizThreadPool = new ThreadPoolExecutor(
                        0,
                        200,
                        60L,
                        TimeUnit.SECONDS,
                        new LinkedBlockingQueue<Runnable>(2000),
                        new ThreadFactory() {
                            @Override
                            public Thread newThread(Runnable r) {
                                return new Thread(r, "xxl-rpc, EmbedServer bizThreadPool-" + r.hashCode());
                            }
                        },
                        new RejectedExecutionHandler() {
                            @Override
                            public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
                                throw new RuntimeException("xxl-job, EmbedServer bizThreadPool is EXHAUSTED!");
                            }
                        });

                // 服务运行端口
                //int port = 50051;
                // 注册对外提供的服务
                server = ServerBuilder.forPort(port)

                        .addService(new ExecutorBizIProtoImpl(executorBiz, accessToken, bizThreadPool))// 注册对外提供的服务
                        .build().start();

                System.out.println("Server started, listening on " + port);

                Runtime.getRuntime().addShutdownHook(new Thread() {
                    @Override
                    public void run() {
                        // 使用标准错误输出，因为日志记录器有可能在JVM关闭时被重置
                        System.err.println("*** shutting down gRPC server since JVM is shutting down");
                        try {
                            GrpcServer.this.stop();
                            stopRegistry();
                        } catch (InterruptedException e) {
                            e.printStackTrace(System.err);
                        }
                        System.err.println("*** server shut down");
                    }
                });

                logger.info(">>>>>>>>>>> xxl-job remoting server start success, nettype = {}, port = {}", GrpcServer.class, port);

                // start registry    执行器注册到admin上
                startRegistry(appname, address);

                blockUntilShutdown();
            } catch (Exception e) {
                if (e instanceof InterruptedException) {
                    logger.info(">>>>>>>>>>> xxl-job remoting server stop.");
                } else {
                    logger.error(">>>>>>>>>>> xxl-job remoting server error.", e);
                }
            } finally {
                // stop
                try {
                    if (!server.isShutdown()){
                        server.shutdownNow();
                        stopRegistry();
                    }
                } catch (Exception e) {
                    logger.error(e.getMessage(), e);
                }
            }
        });

        thread.setDaemon(true);	// daemon, service jvm, user thread leave >>> daemon leave >>> jvm leave
        thread.start();
    }

    // ---------------------- registry ----------------------

    public void startRegistry(final String appname, final String address) {
        // start registry
        ExecutorRegistryThread.getInstance().start(appname, address);
    }

    public void stopRegistry() {
        // stop registry
        ExecutorRegistryThread.getInstance().toStop();
    }





    private void start() throws IOException {

        // 服务运行端口
        int port = 50051;
        // 注册对外提供的服务
        server = ServerBuilder.forPort(port)
                .addService(new ExecutorBizIProtoImpl())// 注册对外提供的服务
                .build().start();

        System.out.println("Server started, listening on " + port);

        Runtime.getRuntime().addShutdownHook(new Thread() {
            @Override
            public void run() {
                // 使用标准错误输出，因为日志记录器有可能在JVM关闭时被重置
                System.err.println("*** shutting down gRPC server since JVM is shutting down");
                try {
                    GrpcServer.this.stop();
                } catch (InterruptedException e) {
                    e.printStackTrace(System.err);
                }
                System.err.println("*** server shut down");
            }
        });
    }

    public void stop() throws InterruptedException {
        if (server != null) {
            server.shutdown().awaitTermination(30, TimeUnit.SECONDS);
        }
    }

    /**
     * 在主线程上等待终止，因为grpc库使用守护进程。
     */
    private void blockUntilShutdown() throws InterruptedException {
        if (server != null) {
            server.awaitTermination();
        }
    }

    /**
     * 启动服务Main方法
     */
    public static void main(String[] args) throws IOException, InterruptedException {
        final GrpcServer server = new GrpcServer();
        server.start();
        server.blockUntilShutdown();
    }

}
